package com.atguigu.flink.chapter5.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.Arrays;
import java.util.List;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/21 11:09
 */
public class Flink03_Sink_ES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<HttpHost> esHosts = Arrays.asList(new HttpHost("hadoop162", 9200), new HttpHost("hadoop163", 9200, "http"), new HttpHost("hadoop164", 9200));
        ElasticsearchSink.Builder<WaterSensor> esSinkBuilder = new ElasticsearchSink.Builder<>(esHosts, new ElasticsearchSinkFunction<WaterSensor>() {
            @Override
            public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) {

                IndexRequest index = Requests
                  .indexRequest("sensor")
                  .type("_doc")
                  .id(element.getId())
                  .source(JSON.toJSONString(element), XContentType.JSON);
                indexer.add(index);

            }
        });
        esSinkBuilder.setBulkFlushInterval(5000);
//        esSinkBuilder.setBulkFlushMaxActions(1);
        env
          .socketTextStream("hadoop162", 9999)
          .map(line -> {
              String[] split = line.split(",");
              return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
          })
          .keyBy(WaterSensor::getId)
          .sum("vc")
          .addSink(esSinkBuilder.build())
          .setParallelism(1);


        env.execute();
    }
}
/*
hash:
key             value(hash)
"sensor"        field           value
                "sensor_1"     {"", "", ..}

 */